Python aiohttp.FormData方法代码示例

2023-02-23 20:51| 来源: 网络整理| 查看: 265

本文整理汇总了Python中aiohttp.FormData方法的典型用法代码示例。如果您正苦于以下问题：Python aiohttp.FormData方法的具体用法？Python aiohttp.FormData怎么用？Python aiohttp.FormData使用的例子？那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在类aiohttp的用法示例。

在下文中一共展示了aiohttp.FormData方法的15个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: send_files # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def send_files(self, channel_id, *, files, content=None, tts=False, embed=None, nonce=None, allowed_mentions=None): r = Route('POST', '/channels/{channel_id}/messages', channel_id=channel_id) form = aiohttp.FormData() payload = {'tts': tts} if content: payload['content'] = content if embed: payload['embed'] = embed if nonce: payload['nonce'] = nonce if allowed_mentions: payload['allowed_mentions'] = allowed_mentions form.add_field('payload_json', utils.to_json(payload)) if len(files) == 1: file = files[0] form.add_field('file', file.fp, filename=file.filename, content_type='application/octet-stream') else: for index, file in enumerate(files): form.add_field('file%s' % index, file.fp, filename=file.filename, content_type='application/octet-stream') return self.request(r, data=form, files=files) 开发者ID:Rapptz，项目名称:discord.py，代码行数:25，代码来源:http.py 示例2: add_formfields # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def add_formfields(self): self.data = aiohttp.FormData() if isinstance(self.data_old, dict): for key, value in self.data_old.items(): self.data.add_field(key, value) elif isinstance(self.data_old, tuple): for td in self.data_old: if isinstance(td, tuple): self.data.add_field(td[0], td[1]) if isinstance(self.files, str): content_type = self.get_content_type(self.files) filename = os.path.basename(self.files) self.data.add_field(filename, open(self.files, 'rb'), content_type=content_type) elif isinstance(self.files, tuple): for file_name in self.files: content_type = self.get_content_type(file_name) filename = os.path.basename(file_name) self.data.add_field(filename, open(file_name, 'rb'), content_type=content_type) elif isinstance(self.files, dict): for file_title, file_name in self.files.items(): content_type = self.get_content_type(file_name) self.data.add_field(file_title, open(file_name, 'rb'), content_type=content_type) 开发者ID:kanishka-linux，项目名称:kawaii-player，代码行数:27，代码来源:req_aio.py 示例3: get_meme # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def get_meme( self, meme: int, boxes: List[Dict[str, str]], username: str, password: str ) -> str: log.debug(boxes) try: form_data = aiohttp.FormData() form_data.add_field("template_id", meme) form_data.add_field("username", username) form_data.add_field("password", password) i = 0 for box in boxes: for k, v in box.items(): form_data.add_field(f"boxes[{i}][{k}]", v) i += 1 async with aiohttp.ClientSession() as session: async with session.post(CAPTION_URL, data=form_data) as r: result = await r.json() if not result["success"]: raise ImgFlipAPIError(result["error_message"]) except Exception as e: log.error("Error grabbing meme", exc_info=True) raise ImgFlipAPIError(e) return result["data"]["url"] 开发者ID:TrustyJAID，项目名称:Trusty-cogs，代码行数:26，代码来源:imgflip.py 示例4: _compose_data # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def _compose_data(req, **user_kw): token, method, params, files = req data = aiohttp.FormData() if params: for key,value in params.items(): data.add_field(key, str(value)) if files: for key,f in files.items(): if isinstance(f, tuple): if len(f) == 2: filename, fileobj = f else: raise ValueError('Tuple must have exactly 2 elements: filename, fileobj') else: filename, fileobj = _guess_filename(f) or key, f data.add_field(key, fileobj, filename=filename) return data 开发者ID:nickoala，项目名称:telepot，代码行数:24，代码来源:api.py 示例5: upload # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def upload(): data = FormData() data.add_field('package', open('tests/QXmokuai_3.apk', 'rb'), filename='QXmokuai_3.apk') data.add_field('msg', 'test upload') _, response = app.test_client.post('/upload/app', data=data) response_normal_check(response) return response 开发者ID:skytoup，项目名称:AppServer，代码行数:9，代码来源:tests.py 示例6: test_allows_post_with_url_encoding # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def test_allows_post_with_url_encoding(client): data = FormData() data.add_field("query", "{test}") response = await client.post( "/graphql", data=data(), headers={"content-type": "application/x-www-form-urlencoded"}, ) assert await response.json() == {"data": {"test": "Hello World"}} assert response.status == 200 开发者ID:graphql-python，项目名称:graphql-server-core，代码行数:13，代码来源:test_graphqlview.py 示例7: request # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def request(session, oauth_token, skill_id=None, method=None, json=None, file=None, request_method='POST', custom_url=None, **kwargs): """ Make a request to API :param session: HTTP Client session :type session: :obj:`aiohttp.ClientSession` :param oauth_token: oauth_token :type oauth_token: :obj:`str` :param skill_id: skill_id. Optional. Not used if custom_url is provided :type skill_id: :obj:`str` :param method: API method. Optional. Not used if custom_url is provided :type method: :obj:`str` :param json: request payload :type json: :obj: `dict` :param file: file :type file: :obj: `io.BytesIO` :param request_method: API request method :type request_method: :obj:`str` :param custom_url: Yandex has very developer UNfriendly API, so some endpoints cannot be achieved by standatd template. :type custom_url: :obj:`str` :return: result :rtype: ::obj:`dict` """ log.debug("Making a `%s` request to %r with json `%r` or file `%r`", request_method, method, json, file) if custom_url is None: url = Methods.api_url(skill_id, method) else: url = custom_url headers = {'Authorization': oauth_token} data = None if file: data = aiohttp.FormData() data.add_field('file', file) try: async with session.request(request_method, url, json=json, data=data, headers=headers, **kwargs) as response: return await _check_result(response) except aiohttp.ClientError as e: raise exceptions.NetworkError(f"aiohttp client throws an error: {e.__class__.__name__}: {e}") 开发者ID:mahenzon，项目名称:aioalice，代码行数:42，代码来源:api.py 示例8: test_predict # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def test_predict(self, api: ApiConfig, client: aiohttp.ClientSession, image_file_obj: io.BytesIO) -> None: predict_url = api.model_base_url + "/predict" data = FormData() data.add_field("file", image_file_obj, filename="test_image.jpg", content_type="image/img") async with client.post(predict_url, data=data) as response: assert response.status == HTTPOk.status_code res_data = await response.json() assert "mean_score" in res_data assert "std_score" in res_data assert "scores" in res_data assert "total_time" in res_data 开发者ID:truskovskiyk，项目名称:nima.pytorch，代码行数:15，代码来源:test_api.py 示例9: put # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def put(session, fid: str, content: bytes): url = f'{seaweedfs_url}/{fid}' data = aiohttp.FormData() data.add_field( 'file', content, content_type='application/gzip' ) async with session.put(url, data=data) as response: result = await response.read() return response.status, result 开发者ID:anyant，项目名称:rssant，代码行数:13，代码来源:benchmark_seaweedfs.py 示例10: upload_audio_message # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def upload_audio_message(api, multipart_data, peer_id): """Upload audio file `multipart_data` and return Attachment for sending to user with id `peer_id`(possibly)""" sender = api.get_default_sender("docs.getMessagesUploadServer") client = api.get_current_sender("docs.getMessagesUploadServer", sender=sender) data = aiohttp.FormData() data.add_field('file', multipart_data, filename="message.mp3", content_type='multipart/form-data') values = {'type': "audio_message", 'peer_id': peer_id} if client.group_id: values['group_id'] = client.group_id response = await api(sender=sender).docs.getMessagesUploadServer(**values) if not response or not response.get('upload_url'): return None upload_url = response['upload_url'] async with aiohttp.ClientSession() as sess: async with sess.post(upload_url, data=data) as resp: result = json.loads(await resp.text()) if not result: return None data = dict(file=result['file']) result = await api(sender=sender).docs.save(**data) if not result: return None return Attachment.from_upload_result(result[0], "doc") 开发者ID:ekonda，项目名称:sketal，代码行数:37，代码来源:helpers.py 示例11: upload_doc # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def upload_doc(api, multipart_data, filename="image.png", additional_params=None): """Upload file `multipart_data` and return Attachment for sending to user""" if additional_params is None: additional_params = {} sender = api.get_default_sender("docs.getWallUploadServer") client = api.get_current_sender("docs.getWallUploadServer", sender=sender) data = aiohttp.FormData() data.add_field('file', multipart_data, filename=filename, content_type='multipart/form-data') values = {} values.update(additional_params) if client.group_id: values['group_id'] = client.group_id response = await api(sender=sender).docs.getWallUploadServer(**values) if not response or not response.get('upload_url'): return None upload_url = response['upload_url'] async with aiohttp.ClientSession() as sess: async with sess.post(upload_url, data=data) as resp: result = json.loads(await resp.text()) if not result or not result.get("file"): print(result) return None data = dict(file=result['file']) result = await api(sender=sender).docs.save(**data) if not result: return None return Attachment.from_upload_result(result[0], "doc") 开发者ID:ekonda，项目名称:sketal，代码行数:43，代码来源:helpers.py 示例12: upload_photo # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def upload_photo(api, multipart_data, peer_id=None): """Upload photo file `multipart_data` and return Attachment for sending to user with id `peer_id`(optional but recommended)""" sender = api.get_default_sender('photos.getMessagesUploadServer') data = aiohttp.FormData() data.add_field('photo', multipart_data, filename='picture.png', content_type='multipart/form-data') if peer_id: kwargs = {"peer_id": peer_id} else: kwargs = {} response = await api(sender=sender).photos.getMessagesUploadServer(**kwargs) if not response or not response.get('upload_url'): return None upload_url = response['upload_url'] async with aiohttp.ClientSession() as sess: async with sess.post(upload_url, data=data) as resp: result = json.loads(await resp.text()) if not result: return None result = await api(sender=sender).photos.saveMessagesPhoto( **{'photo': result['photo'], 'hash': result['hash'], 'server': result['server']}) if not result: return None return Attachment.from_upload_result(result[0]) 开发者ID:ekonda，项目名称:sketal，代码行数:38，代码来源:helpers.py 示例13: do_retro # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def do_retro(self, text, bcg): if '|' not in text: if len(text) >= 15: text = [text[i:i + 15] for i in range(0, len(text), 15)] else: split = text.split() if len(split) == 1: text = [x for x in text] if len(text) == 4: text[2] = text[2]+text[-1] del text[3] else: text = split else: text = text.split('|') headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64; rv:43.0) Gecko/20100101 Firefox/43.0'} payload = aiohttp.FormData() payload.add_field('current-category', 'all_effects') payload.add_field('bcg', bcg) payload.add_field('txt', '4') count = 1 for s in text: if count > 3: break payload.add_field('text'+str(count), s.replace("'", "\'")) count += 1 try: with aiohttp.ClientSession() as session: with aiohttp.Timeout(5): async with session.post('https://photofunia.com/effects/retro-wave?server=3', data=payload, headers=headers) as r: txt = await r.text() except asyncio.TimeoutError: return match = self.retro_regex.findall(txt) if match: download_url = match[0][0] b = await self.bytes_download(download_url) return b return False 开发者ID:NotSoSuper，项目名称:NotSoBot，代码行数:41，代码来源:Fun.py 示例14: upload_log # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def upload_log(self, file, user): params = {"json": 1} token = await self.get_dpsreport_usertoken(user) if token: params["userToken"] = token data = aiohttp.FormData() data.add_field("file", await file.read(), filename=file.filename) async with self.session.post( UPLOAD_URL, data=data, params=params) as r: resp = await r.json() error = resp["error"] if error: raise APIError(error) return resp 开发者ID:Maselkov，项目名称:GW2Bot，代码行数:16，代码来源:evtc.py 示例15: make_mulitpart_form # 需要导入模块: import aiohttp [as 别名] # 或者: from aiohttp import FormData [as 别名] def make_mulitpart_form(self, fields, aio=False): """Create a multipart form to be used across the Symphony API, that works for both requests and the asynchronous aiohttp. Requests basically uses requests-toolbelt, but it's a little bit more involved for aiohttp. The output of this is expected to be passed to either execute_rest_request or execute_rest_request_async depending whether aio was true""" if aio: # This appears to be the canonical way to use aiohttp to pass mulipart data into the API # in the same way that MultipartEncoder does for Requests. # aiohttp.FormData does appear to work because of the way the Symphony API demands a boundary # in the header. aiohttp.MultipartWriter.append_form doesn't appear to work because it # encodes as a application/x-www-form-urlencoded that Symphony doesn't appear to like for # attachments with aiohttp.MultipartWriter("form-data") as data: for k, v in fields.items(): if len(v) == 1: part = data.append(v) part.set_content_disposition("form-data", name=k) if len(v) == 3: filename, file_object, content_type = v part = data.append(file_object, {'Content-Type': content_type}) part.set_content_disposition('form-data', name=k, filename=filename) headers = { 'Content-Type': 'multipart/form-data; boundary=' + data.boundary } else: print(fields) data = MultipartEncoder( fields=fields ) headers = { 'Content-Type': data.content_type } return {"data": data, "headers": headers} 开发者ID:SymphonyPlatformSolutions，项目名称:symphony-api-client-python，代码行数:39，代码来源:api_client.py

注：本文中的aiohttp.FormData方法示例由纯净天空整理自Github/MSDocs等开源代码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。

【本文地址】

公司简介

联系我们

今日新闻

推荐新闻

专题文章